package com.snopy.mq.kafka;

import com.alibaba.fastjson.JSON;
import com.snopy.pojo.Banana;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author snopy
 * @version 1.0.0
 * @classname KProducer
 * @description
 * @email 77912204@qq.com
 * @date 2022/06/08 14:03
 */
public class KProducer {
    public static void main(String[] args) {
        //System.out.println(StringSerializer.class.getName());
        //1.配置生产者启动参数
        Properties props = new Properties();
        props.put("bootstrap.servers","192.168.164.101:9092");
        //发送前等待的时间 ProducerConfig.LINGER_MS_CONFIG 默认情况下，缓冲区不满，也是立即发送的，如果想要减少发送请求的次数，
        // 可以设置lingger.ms的时间大于0，那么就会再等待lingger.ms后再发送
        props.put(ProducerConfig.LINGER_MS_CONFIG,1);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());

        props.put(ProducerConfig.BATCH_SIZE_CONFIG,1);
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        /*
        * acks设置在于主副本的消息写入确认，设置为0时，消息放入缓冲区后就认为消息已经发送，然后立即返回结果了。这个时候retries是无效的。因为不知道是否真的发送成功。
        * acks设置为1的时候，当主本写入确认后，立即返回，不会等待副本写入的确认。如果在主本确认后，副本写入前失败了，那么消息就会丢失。
        * acks设置为all或者-1，就是当主副本都写入确认之后，再返回确认结果。
        * */
        props.put(ProducerConfig.ACKS_CONFIG,-1);
        //max.block.ms 最大阻塞时间，当缓冲区的record发送到消息服务器的速度比record记录发送到缓冲区要快时，缓冲区就会被填满，这时就会阻塞，

        //kafka从0.11版本开始，支持幂等生产者和事务生产者，
        //从3.0版本开始，提供了enable.idempotence配置，控制发送消息的幂等性，默认为true,
        //如果enable.idepotence为true时，retries默认值为Integer.MAX_VALUE. acks的配置为all。这个api的变更，所以已存在的系统无需修改就可以使用这个功能。
        //2.传递kafka启动参数，创建生产者
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(props);
        Banana banana = new Banana("green","raw banana ",10.5,"this is a banana!");
        //3.构建消息内容
        ProducerRecord record = new ProducerRecord("test", JSON.toJSONString(banana));
        //4.传递消息
        Future<Metadata> send = producer.send(record);
        try {
            Metadata metadata = send.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {

            e.printStackTrace();
        }
        producer.close();

    }
}
